/**
 *
 */
package com.rrd.hive.udaf;

import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.hive.ql.exec.UDFArgumentTypeException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.parse.SemanticException;
import org.apache.hadoop.hive.ql.udf.generic.AbstractGenericUDAFResolver;
import org.apache.hadoop.hive.ql.udf.generic.GenericUDAFEvaluator;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.StandardListObjectInspector;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorUtils;
import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
import org.apache.hadoop.io.Text;

/**
 * @author liubaoxin
 * @version 创建时间：2018年01月23日
 *          计算进件最近24个月的还款状态
 */
public class CreditRepayState24UDAF extends AbstractGenericUDAFResolver {

    @Override
    public GenericUDAFEvaluator getEvaluator(TypeInfo[] parameters)
            throws SemanticException {
        if (parameters.length >= 4) {
            return new FilterConcatEvaluator();

        } else {
            throw new UDFArgumentTypeException(parameters.length, "输入参数不正确!");
        }

    }

    public static class FilterConcatEvaluator extends GenericUDAFEvaluator {
        private PrimitiveObjectInspector inputOI; // 应还日期、逾期天数、还清时间
        private PrimitiveObjectInspector inputOI2;  //上报年月
        private PrimitiveObjectInspector inputOI3;  //结清年月
        private StandardListObjectInspector listOI;
        private Text result = new Text();

        @Override
        public ObjectInspector init(Mode m, ObjectInspector[] parameters)
                throws HiveException {
            super.init(m, parameters);

            if (m == Mode.PARTIAL1) {
                this.inputOI = (PrimitiveObjectInspector) parameters[0];
                this.inputOI2 = (PrimitiveObjectInspector) parameters[1];

                return ObjectInspectorFactory.getStandardListObjectInspector(
                    PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.PARTIAL2) {
                this.listOI = (StandardListObjectInspector) parameters[0];
                return ObjectInspectorFactory.getStandardListObjectInspector(
                    PrimitiveObjectInspectorFactory.javaStringObjectInspector);
            } else if (m == Mode.FINAL) {
                this.listOI = (StandardListObjectInspector) parameters[0];
                return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
            } else if (m == Mode.COMPLETE) {
                return PrimitiveObjectInspectorFactory.javaStringObjectInspector;
            } else { //
                throw new RuntimeException("no such mode Exception");
            }
        }

        static class ResultAgg implements AggregationBuffer {
            String cur_ym = "";
            String payoff = "";
            List<String> list = new ArrayList<>();
        }

        @Override
        public AggregationBuffer getNewAggregationBuffer()
                throws HiveException {
            // TODO Auto-generated method stub
            AggregationBuffer aggr = new ResultAgg();
            this.reset(aggr);
            return aggr;
        }

        @Override
        public void reset(AggregationBuffer agg) throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg obj = (ResultAgg) agg;
            obj.list = new ArrayList<>();
            obj.cur_ym = "";
            obj.payoff = "";

        }

        @Override
        public void iterate(AggregationBuffer agg, Object[] parameters)
                throws HiveException {
            // TODO Auto-generated method stub
            if (parameters == null || parameters[0] == null) {
                return;
            }

            ResultAgg rag = (ResultAgg) agg;

            rag.cur_ym = PrimitiveObjectInspectorUtils.getString(parameters[0],
                this.inputOI);
            rag.list.add(PrimitiveObjectInspectorUtils.getString(parameters[1],
                this.inputOI2));
            rag.list.add(PrimitiveObjectInspectorUtils.getString(parameters[2],
                this.inputOI3));

        }

        @Override
        public Object terminatePartial(AggregationBuffer agg)
                throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg myagg = (ResultAgg) agg;
            List<String> ret = new ArrayList<>(myagg.list);

            return ret;
        }

        @Override
        public Object terminate(AggregationBuffer agg) throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg aggr = (ResultAgg) agg;
            if (aggr.list == null || aggr.list.size() == 0) {
                // System.out.println("result is null");
                this.result.set("\\\\\\\\\\\\\\\\\\\\\\\\");
            } else {
                this.result
                    .set(this.to_status(aggr.list, aggr.cur_ym, aggr.payoff));
            }

            return this.result;
        }

        private String to_status(List<String> list, String ym, String payoff) {

            StringBuilder sb = new StringBuilder("*");
            StringBuilder sb2 = new StringBuilder("");
            String due_ym = "";
            payoff = payoff == null || payoff.equals("") ? "417201" : payoff;
            for (String s : list) {
                due_ym = s.substring(0, 6);
                if (due_ym.compareTo(ym) <= 0 && due_ym.compareTo(payoff) < 0) {
                    sb.append(this.getStatus(Integer.parseInt(s.substring(6))));
                }
            }
            if (!payoff.equals("417201")) {
                sb.append("C");
            }
            for (int i = sb.length(); i < 24; i++) {
                sb2.append("/");
            }
            sb2.append(sb);
            return sb2.toString();

        }

        private String getStatus(int n) {
            String s = "";
            if (n >= 1 && n <= 30) {
                s = "1";
            } else if (n >= 31 && n <= 60) {
                s = "2";
            } else if (n >= 61 && n <= 90) {
                s = "3";
            } else if (n >= 91 && n <= 120) {
                s = "4";
            } else if (n >= 121 && n <= 150) {
                s = "5";
            } else if (n >= 151 && n <= 180) {
                s = "6";
            } else if (n >= 181) {
                s = "7";
            } else {
                s = "N";
            }
            return s;
        }

        @Override
        public void merge(AggregationBuffer agg, Object partial)
                throws HiveException {
            // TODO Auto-generated method stub
            ResultAgg aggr = (ResultAgg) agg;
            if (partial != null) { // 在没有输入数据情况下，null会传进来
                List partialResult = this.listOI.getList(partial);
                for (Object key : partialResult) {
                    aggr.list.add((String) key);

                }
            }
        }
    }

}
